package belf.migrate.plugin.mysql;

import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.*;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.plugin.mysql.antlr.MySqlLexer;
import belf.migrate.plugin.mysql.antlr.MySqlParser;
import org.antlr.v4.runtime.CharStream;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.ParseTreeWalker;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.plugin.mysql.listener.MySql2DMFunctionListener;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import belf.migrate.api.util.ExceptionUtils;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;


import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

@Slf4j
public class MySQLCatalog extends AbstractJdbcCatalog {
    private static final String QUOTE = "`";

    static {
        SYS_DATABASES.addAll(Collections.unmodifiableList(
                Arrays.asList("INFORMATION_SCHEMA", "MYSQL", "SYSSSO", "PERFORMANCE_SCHEMA", "SYS")));
    }

    public MySQLCatalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.MYSQL, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getCreateTableSql(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        throw new UnsupportedOperationException();
    }

    @Override
    public String getExistDataSql(TablePath tablePath) {
        return String.format(
                "SELECT * FROM %s.%s LIMIT 1;",
                tablePath.getRightDatabaseName(), tablePath.getTableName());
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return String.format(
                "SELECT count(*) FROM %s.%s LIMIT 1;",
                tablePath.getRightDatabaseName(), tablePath.getTableName());
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl;
        if (StringUtils.isNotBlank(tablePath.getDatabaseName())) {
            dbUrl = getUrlFromDatabaseName(tablePath.getDatabaseName());
        } else {
            dbUrl = getUrlFromDatabaseName(defaultDatabase);
        }


        String query = "SELECT NON_UNIQUE, INDEX_NAME, SEQ_IN_INDEX,COLUMN_NAME,COLLATION,INDEX_TYPE FROM INFORMATION_SCHEMA.STATISTICS " +
                "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? ";

        try {
            Connection conn = getConnection(dbUrl);
            PreparedStatement pstmt = conn.prepareStatement(query);

            DatabaseMetaData metaData = conn.getMetaData();
            List<IndexKey> defaultIndexKeys =  CatalogUtils.getIndexKeys(metaData, tablePath);

            // 补充获取mysql索引的额外信息
            pstmt.setString(1, tablePath.getDatabaseName());
            pstmt.setString(2, tablePath.getTableName());
            // index name -> index
            Map<String, IndexKeyExt> indexKeyMap = new HashMap<>();

            try (ResultSet rs = pstmt.executeQuery()) {
                log.debug("query index:{}", query);
                while (rs.next()) {
                    boolean unique = rs.getInt("NON_UNIQUE")==0?true:false;
                    String indexName = rs.getString("INDEX_NAME");
                    int sequenceInIndex = rs.getInt("SEQ_IN_INDEX");
                    String columnName = rs.getString("COLUMN_NAME");
                    String indexOrder = rs.getString("COLLATION");
                    String indexTypeName = rs.getString("INDEX_TYPE");
                    List<IndexKeyExt.ColumnWithSequence> columns = new ArrayList<>();
                    columns.add(new IndexKeyExt.ColumnWithSequence(columnName, sequenceInIndex));
                    IndexKeyExt IndexKeyExt = new IndexKeyExt(
                            columns,
                            indexName,
                            unique,
                            indexOrder,
                            indexTypeName);
                    IndexKeyExt existOne = indexKeyMap.get(indexName);
                    if (existOne != null) {
                        existOne.getColumns().addAll(IndexKeyExt.getColumns());
                    } else {
                        indexKeyMap.put(indexName, IndexKeyExt);
                    }
                    log.debug("got index: {}", IndexKeyExt);
                }
            }

            defaultIndexKeys.forEach(indexKey -> {
                IndexKeyExt IndexKeyExt = indexKeyMap.get(indexKey.getIndexName());
                if (null != IndexKeyExt) {
                    indexKey.setColumnNames(IndexKeyExt.getSortedColumnNames());
                    indexKey.setIndexTypeName(IndexKeyExt.getIndexTypeName());
                }
            });
            return defaultIndexKeys;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<String> getCreateIndexSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        throw new UnsupportedOperationException();
    }

    @Override
    public String getListDatabaseSQL() {
        return "SHOW DATABASES;";
    }

    @Override
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
        String sql = """
                SELECT COLUMN_NAME,COLUMN_DEFAULT,EXTRA,COLUMN_COMMENT,DATETIME_PRECISION FROM information_schema.columns\s
                WHERE table_schema = ? AND table_name = ?
                """;
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ps.setString(1, tablePath.getDatabaseName());
            ps.setString(2, tablePath.getTableName());

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                String columnDefault = rs.getString("COLUMN_DEFAULT");
                String extra = rs.getString("EXTRA");
                String EXTRA_PREFIX = "DEFAULT_GENERATED";
                if (extra.toUpperCase().startsWith(EXTRA_PREFIX)) {
                    extra = extra.substring(EXTRA_PREFIX.length()).trim();
                }
                String comment = rs.getString("COLUMN_COMMENT");
                int dateTimePrecision = rs.getInt("DATETIME_PRECISION");
                Optional<ColumnMetaData> possibleColumn = columns.stream().filter(new Predicate<ColumnMetaData>() {
                    @Override
                    public boolean test(ColumnMetaData columnMetaData) {
                        return columnMetaData.getColumnName().equalsIgnoreCase(columnName);
                    }
                }).findFirst();
                if (possibleColumn.isPresent()) {
                    ColumnMetaData columnMetaData = possibleColumn.get();
                    ColumnExtra columnExtra = new ColumnExtra(columnDefault, extra, comment, dateTimePrecision);
                    columnMetaData.setExtra(columnExtra);
                }
            }
        } catch (SQLException e) {
            log.error("Error on query mysql column extra information:\n{}", ExceptionUtils.toString(e));
        }
    }

    /**
     * mysql查询指定数据库下的所有视图
     * @param databaseName
     * @return
     */
    @Override
    public String getListViewsSQL(String databaseName) {
        return String.format("SELECT TABLE_NAME AS VIEW_NAME FROM INFORMATION_SCHEMA.VIEWS WHERE TABLE_SCHEMA = '%s'", databaseName);
    }

    @Override
    public String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return String.format("SELECT TABLE_NAME as VIEW_NAME FROM INFORMATION_SCHEMA.VIEWS WHERE TABLE_SCHEMA = '%s'", defaultDatabase);
    }

    @Override
    public String getListProcedureSQL(String databaseName) {
        return String.format("SELECT ROUTINE_NAME AS PROCEDURE_NAME FROM information_schema.ROUTINES\n" +
                "WHERE ROUTINE_SCHEMA='%s' AND ROUTINE_TYPE = 'PROCEDURE'\n" +
                "ORDER BY ROUTINE_NAME", databaseName);
    }

    @Override
    public String getListTriggerSQL(String databaseName) {
        return String.format("SELECT TRIGGER_NAME FROM INFORMATION_SCHEMA.TRIGGERS\n" +
                "WHERE TRIGGER_SCHEMA = '%s'", databaseName);
    }

    @Override
    public String getListFunctionSQL(String databaseName) {
        return String.format("""
                SELECT ROUTINE_NAME as FUNCTION_NAME
                FROM INFORMATION_SCHEMA.ROUTINES 
                WHERE ROUTINE_TYPE = 'FUNCTION' 
                AND ROUTINE_SCHEMA = '%s'
                """, databaseName);
    }

    @Override
    public String getListTableSQL(String database, String schema) {
        return String.format("SELECT TABLE_NAME AS TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%s' AND TABLE_TYPE = 'BASE TABLE'", database);
    }

    @Override
    protected String getListIndexSQL(String defaultDatabase) {
        return String.format("SELECT INDEX_NAME,TABLE_NAME,INDEX_TYPE FROM information_schema.STATISTICS WHERE TABLE_SCHEMA='%s' ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX", defaultDatabase);
    }

    @Override
    public List<UserDefFunction> listFunctions(String schema, List<String> functionNames) throws CatalogException {
        List<String> limitedFunctionNames = functionNames;
        if (null == functionNames || functionNames.isEmpty()) {
            limitedFunctionNames = listFunctionNames(schema, null);
        }

        List<UserDefFunction> userDefFunctions = new ArrayList<>();
        try (Connection connection = getConnection(getUrlFromDatabaseName(schema))){
            for (String functionName : limitedFunctionNames) {
                String showDdl = "SHOW CREATE FUNCTION " + functionName;
                Statement statement = connection.createStatement();
                ResultSet rs = statement.executeQuery(showDdl);
                if (rs.next()) {
                    String ddl = rs.getString("Create Function");
                    userDefFunctions.add(new UserDefFunction(functionName, ddl));
                }
                rs.close();
                statement.close();
            }
        } catch (SQLException e) {
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_DDL_ERROR, "查询mysql函数的DDL失败", e);
        }
        return userDefFunctions;
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName) throws CatalogException, DatabaseNotExistException {
        List<TableInfoModel> tables = new ArrayList<>();
        try {
            String sql = "SELECT TABLE_NAME, TABLE_ROWS, TABLE_COMMENT, TRUNCATE(data_length/1024, 2) AS TABLE_SIZE FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'";
            log.info("Query mysql all tables: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            preparedStatement.setString(1, databaseName);
            ResultSet resultSet = preparedStatement.executeQuery();
            while (resultSet.next()) {
                TableInfoModel table = new TableInfoModel();
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setComment(resultSet.getString("TABLE_COMMENT"));
                table.setSize(resultSet.getDouble("TABLE_SIZE"));
                tables.add(table);
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return tables;
    }

    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) {
        TableInfoModel table = new TableInfoModel();
        try {
            String sql = "SELECT TABLE_NAME, TABLE_ROWS, TABLE_COMMENT, TRUNCATE(data_length/(1024 * 1024), 2) AS TABLE_SIZE FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE' AND TABLE_NAME = ?";
            log.info("Query mysql table info: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            preparedStatement.setString(1, tablePath.getDatabaseName());
            preparedStatement.setString(2, tablePath.getTableName());
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setComment(resultSet.getString("TABLE_COMMENT"));
                table.setSize(resultSet.getDouble("TABLE_SIZE"));
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("获取表的信息失败！", e);
        }
        return table;
    }

    /**
     * 显示MySQL创建表的语句。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        String sql = "SHOW CREATE TABLE " + tablePath.getDatabaseAndTableName();
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("Create Table");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new TableNotExistException(DatabaseIdentifier.MYSQL, tablePath);
        }
        return null;
    }

    /**
     * 显示MySQL创建视图的语句。注意，对于"SHOW CREATE VIEW "这种执行，不能使用PreparedStatement预处理语句，
     * 只能使用 Statement。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {
        String sql = "SHOW CREATE VIEW " + tablePath.getDatabaseAndTableName();
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("Create View");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new ViewNotExistException(DatabaseIdentifier.MYSQL, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        String sql = "SHOW CREATE PROCEDURE " + tablePath.getDatabaseAndTableName();
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("Create Procedure");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new ProcedureNotExistException(DatabaseIdentifier.MYSQL, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException {
        String sql = "SHOW CREATE TRIGGER " + tablePath.getDatabaseAndTableName();
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("SQL Original Statement");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        String sql = "SHOW CREATE FUNCTION " + tablePath.getDatabaseAndTableName();
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("Create Function");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            throw new FunctionNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
        return null;
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getDatabaseAndTableName();
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    @Override
    public List<String> specialColumns() {
        return List.of("ENUM", "SET", "STRUCT", "ARRAY", "GEOMETRY", "POINT",
                "LINESTRING", "POLYGON", "MULTIPOINT", "MULTILINESTRING", "MULTIPOLYGON", "GEOMETRYCOLLECTION");
    }

    @Override
    public List<String> bigDataColumns() {
        return List.of("BLOB", "LONGBLOB", "MEDIUMBLOB");
    }

    private final static String QUERY_UNIQUE = """
            SELECT
                TABLE_NAME,
                COLUMN_NAME,
                CONSTRAINT_NAME
            FROM
                information_schema.KEY_COLUMN_USAGE
            WHERE
                TABLE_SCHEMA = ?
                AND TABLE_NAME = ?
                AND CONSTRAINT_NAME IN (
                    SELECT CONSTRAINT_NAME
                    FROM information_schema.TABLE_CONSTRAINTS
                    WHERE CONSTRAINT_TYPE = 'UNIQUE'
                    AND TABLE_SCHEMA = ?
                );
            """;
    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(QUERY_UNIQUE)) {
            ps.setString(1, tablePath.getDatabaseName());
            ps.setString(2, tablePath.getTableName());
            ps.setString(3, tablePath.getDatabaseName());
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String column = rs.getString("COLUMN_NAME");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    private final static String QUERY_CHECK_CONSTRAINT = """
            SELECT
                tc.CONSTRAINT_NAME AS CONSTRAINT_NAME,
                cc.CHECK_CLAUSE AS SEARCH_CONDITION
            FROM
                information_schema.TABLE_CONSTRAINTS tc
            JOIN
                information_schema.CHECK_CONSTRAINTS cc
            ON
                tc.CONSTRAINT_NAME = cc.CONSTRAINT_NAME
            WHERE
                tc.CONSTRAINT_TYPE = 'CHECK'
                AND tc.TABLE_SCHEMA = ?
                AND tc.TABLE_NAME = ?
            """;
    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(QUERY_CHECK_CONSTRAINT)) {
            ps.setString(1, tablePath.getDatabaseName());
            ps.setString(2, tablePath.getTableName());
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    checkConstraints.add(new CheckConstraint(tablePath.getTableName(),
                            rs.getString("CONSTRAINT_NAME"),
                            rs.getString("SEARCH_CONDITION")));
                }
            }

        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Check约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }

    @Override
    protected String getColumnsSql(TablePath tablePath) {
        String sql = "SELECT a.ordinal_position AS ordinal_position, " +
                "a.TABLE_CATALOG as catalog_name, " +
                "a.TABLE_SCHEMA as schema_name," +
                "a.table_name AS table_name, " +
                "a.column_name AS column_name, " +
                "a.data_type AS column_type, " +
                "  CASE  " +
                "  WHEN a.is_nullable = 'YES' THEN 1 " +
                "  ELSE 0 " +
                " END AS nullable, " +
                "a.column_default AS data_default, " +
                "a.column_comment AS column_comment, " +
                "a.column_key AS column_key " +
                " , CASE  " +
                "  WHEN a.column_key = 'PRI' THEN 1 " +
                "  ELSE 0 " +
                " END AS is_primary_key " +
                " , CASE  " +
                "  WHEN extra = 'auto_increment' THEN true " +
                "  ELSE false " +
                " END AS is_increment, a.CHARACTER_MAXIMUM_LENGTH AS data_length " +
                "FROM information_schema.COLUMNS a " +
                "WHERE a.table_schema = '" + tablePath.getRightDatabaseName() + "' ";
        if (StringUtils.isNotEmpty(tablePath.getTableName())) {
            sql = sql + " AND a.Table_Name in ('" + tablePath.getTableName() + "') ";
        }
        sql = sql + "ORDER BY ordinal_position ASC";
        return sql;
    }

    @Override
    protected String getDataSqlNoPk(TablePath tablePath, int limit, int offset) {
        return String.format("SELECT * FROM %s limit %s,%s", getFullTableName(tablePath), limit, offset);
    }

    @Override
    public List<String> findFunctionIncompatible(String sql, String sourceCatalog, String sinkCatalog) {
        CharStream input = CharStreams.fromString(sql);

        // 创建词法分析器和解析器
        MySqlLexer lexer = new MySqlLexer(input);
        CommonTokenStream tokens = new CommonTokenStream(lexer);
        MySqlParser parser = new MySqlParser(tokens);

        // 从解析器中获取起始规则并进行解析
        MySqlParser.RootContext tree = parser.root();

        // 创建ParseTreeWalker
        ParseTreeWalker walker = new ParseTreeWalker();

        // 使用自定义Listener遍历解析树
        MySql2DMFunctionListener listener = new MySql2DMFunctionListener(sourceCatalog, sinkCatalog);
        walker.walk(listener, tree);
        return listener.getDifferentFunctions();
    }

    @Override
    public Object getColumnValueByType(ColumnMetaData columnMetaData, ResultSet resultSet, int index) throws SQLException {
        if ("YEAR".equalsIgnoreCase(columnMetaData.getColumnTypeName())) {
            return resultSet.getInt(index);
        }
        if("TINYINT".equalsIgnoreCase(columnMetaData.getColumnTypeName())) {
            return resultSet.getInt(index);
        }
        return super.getColumnValueByType(columnMetaData, resultSet, index);
    }

    @Override
    public void setCurrentDatabase(Connection connection, String database) throws SQLException {
        String sql = "USE " + database;
        connection.createStatement().execute(sql);
    }


    private MySqlVersion resolveVersion() {
        try (Statement statement = getConnection(defaultUrl).createStatement();
             ResultSet resultSet = statement.executeQuery("SELECT VERSION()")) {
            resultSet.next();
            return MySqlVersion.parse(resultSet.getString(1));
        } catch (Exception e) {
            log.info(
                    "Failed to get mysql version, fallback to default version: {}",
                    MySqlVersion.V_5_7,
                    e);
            return MySqlVersion.V_5_7;
        }
    }

    @Override
    public boolean isAuditLog() {
        try (Statement statement = getConnection(defaultUrl).createStatement();
             ResultSet resultSet = statement.executeQuery("SHOW VARIABLES LIKE 'audit_log_enabled'")) {
            while (resultSet.next()) {
                if ("ON".equalsIgnoreCase(resultSet.getString(2))) {
                    return true;
                }
            }
        } catch (Exception e) {
            log.info("Failed to get mysql audit log status, fallback to false", e);
        }
        return false;
    }

    @Override
    public String[] getScriptDelimiters() {
        return SQLConstants.DEFAULT_SCRIPT_DELIMITER; //$NON-NLS-1$
    }

    @Override
    public String[][] getBlockBoundStrings() {
        // No anonymous blocks in MySQL
        return null;
    }
}
